package org.springblade.mqtt;

import lombok.Data;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springblade.EquipmentApplication;
import org.springblade.common.config.MqQueueConfig;
import org.springblade.common.config.MqttConfig;
import org.springblade.common.tool.ThreadPoolUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.CompletableFuture;

/**
 * 初始化MQTT客户端
 * @author 李家民
 */
@Component
@Data
public class MqttInitialized {

	@Autowired
	public ApplicationContext applicationContext;
	
	/** MQTT客户端 */
	private static MqttClient client = null;
	/** 连接选项 */
	private static MqttConnectOptions connOpts = null;
	/** 连接状态 */
	private static Boolean connectStatus = false;

	public static MqttClient getClient() {
		return client;
	}

	public static void setConnectStatus(Boolean connectStatus) {
		MqttInitialized.connectStatus = connectStatus;
	}

	@Resource
	private MqttConfig mqttConfig;

	/**
	 * MQTT客户端启动
	 */
	@PostConstruct
	private void connect() {
		try {
			// MQTT 连接选项
			connOpts = new MqttConnectOptions();
			connOpts.setUserName(mqttConfig.getUSERNAME());
			connOpts.setPassword(mqttConfig.getPASSWD().toCharArray());
			connOpts.setAutomaticReconnect(true);
			connOpts.setCleanSession(true);
			//  持久化
			MemoryPersistence persistence = new MemoryPersistence();
			// MQ客户端建立
			client = new MqttClient(mqttConfig.getIP_ADDRESS(), applicationContext.getId(), persistence);
			// 设置回调
			client.setCallback(new MqttHandle());
			// 建立连接
			client.connect(connOpts);
			connectStatus = client.isConnected();
			// 监听循环
			CompletableFuture.runAsync(() -> {
				try {
					while (true) {
						Thread.sleep(30000);
						if (!connectStatus) {
							System.out.println("MQTT服务器连接失败!!开始进行重新连接....");
							client.connect(connOpts);
							connectStatus = client.isConnected();
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}, ThreadPoolUtils.getThreadPool());

		} catch (Exception e) {
			connectStatus = client.isConnected();
			e.printStackTrace();
		}
	}


}
